Atom AI Labs - AI-Powered Multi-Tenant Platform

Agent Integration Guide

**Last Updated:** 2026-03-28

**Phase:** 204 - Agent Step Integration

**Component:** Agent-Triggered Integration Operations

Overview

AI agents can trigger native integration operations (Slack, Salesforce, HubSpot, etc.) through the EntitySkillExecutor with full governance enforcement. This guide explains how agents interact with integrations, maturity-based permissions, workflow patterns, and multi-tenant security.

**Key Architecture:**

**EntitySkillExecutor:** Central executor for integration operations with governance
**IntegrationRegistry:** Service discovery and credential resolution (Phase 201)
**AgentGovernanceService:** Maturity-based permission enforcement
**ProposalService:** Intern agent approval workflows
**WorldModel:** Agent learning from execution outcomes

**Critical Principles:**

NO LangChain - Uses EntitySkillExecutor patterns with existing services
Multi-tenant isolation enforced at database layer
All operations checked via AgentGovernanceService
Audit trail maintained for compliance

Agent Maturity Levels

Student Agents

**Can do:** Read operations only (search, list, get, fetch)

**Cannot do:** Write operations (create, update, delete, send)

**Behavior:** Returns errors for destructive operations

**Example operations:**

get_channels (Slack)
get_account (Salesforce)
get_contact (HubSpot)
list_tickets (Zendesk)

**Governance:** Level 1 operations only

Intern Agents

**Can do:** Read operations, draft proposals

**Cannot do:** Direct write operations

**Behavior:** Creates proposals for autonomous supervisor review

**Example operations:**

All read operations (Level 1)
Draft proposals for write operations (Level 2+)

**Workflow:**

Intern agent attempts write operation
EntitySkillExecutor detects intern maturity
Proposal created via ProposalService
Workflow pauses with status: "proposal_created"
Autonomous supervisor approves
Operation executes or workflow resumes

**Governance:** Level 1-2 operations direct, Level 3-4 require proposals

Supervised Agents

**Can do:** Read and write operations (create, update, send, post)

**Cannot do:** Critical operations (delete, bulk operations)

**Behavior:** Executes with live monitoring

**Example operations:**

All read operations (Level 1)
send_message (Slack)
create_lead (Salesforce)
create_deal (HubSpot)
update_contact (HubSpot)

**Governance:** Level 1-3 operations

Autonomous Agents

**Can do:** All operations (read, write, delete, bulk)

**Behavior:** Executes with full autonomy, can supervise interns

**Example operations:**

All operations (Level 1-4)
delete_message (Slack)
delete_lead (Salesforce)
bulk_update (HubSpot)

**Supervision:** Can approve intern proposals

**Governance:** All operations

Integration Operations

Available Operations

Each integration exposes operations with complexity levels:

Slack

Operation	Complexity	Maturity Required	Description
`get_channels`	1	Student	List Slack channels
`get_messages`	1	Student	Retrieve message history
`send_message`	3	Supervised	Send message to channel
`delete_message`	4	Autonomous	Delete message

Salesforce

Operation	Complexity	Maturity Required	Description
`get_account`	1	Student	Retrieve account details
`get_lead`	1	Student	Retrieve lead details
`create_lead`	3	Supervised	Create new lead
`update_lead`	3	Supervised	Update existing lead
`delete_lead`	4	Autonomous	Delete lead

HubSpot

Operation	Complexity	Maturity Required	Description
`get_contact`	1	Student	Retrieve contact details
`get_deal`	1	Student	Retrieve deal details
`create_contact`	3	Supervised	Create new contact
`create_deal`	3	Supervised	Create new deal
`update_deal`	3	Supervised	Update deal
`delete_contact`	4	Autonomous	Delete contact

Asana

Operation	Complexity	Maturity Required	Description
`get_task`	1	Student	Retrieve task details
`get_project`	1	Student	Retrieve project details
`create_task`	3	Supervised	Create new task
`update_task`	3	Supervised	Update task
`delete_task`	4	Autonomous	Delete task

Gmail

Operation	Complexity	Maturity Required	Description
`get_message`	1	Student	Retrieve email
`list_messages`	1	Student	List emails
`send_email`	3	Supervised	Send email
`delete_message`	4	Autonomous	Delete email

Operation Complexity Levels

**Level 1: Read-Only (Student)**

Operations: get, list, search, fetch, retrieve
Examples: get_account, list_messages, search_contacts
Risk: No data modification

**Level 2: Propose/Draft (Intern)**

Operations: draft, suggest, analyze, recommend
Examples: draft_email, suggest_response, analyze_data
Risk: No direct execution

**Level 3: Execute (Supervised)**

Operations: create, update, send, post, add
Examples: create_lead, send_message, update_contact
Risk: Data modification, reversible

**Level 4: Critical (Autonomous)**

Operations: delete, execute, deploy, transfer, bulk_*
Examples: delete_lead, execute_playbook, bulk_update
Risk: Destructive, hard to reverse

Multi-Step Workflows

Workflow Definition

Workflows chain multiple integration operations with result passing:

workflow_steps = [
    {
        "connector": "salesforce",
        "operation": "get_account",
        "args": {"id": "001xx000003"}
    },
    {
        "connector": "slack",
        "operation": "send_message",
        "args": {
            "channel": "#sales",
            "text": "Account updated: ${step_1.name} (${step_1.billing_state})"
        }
    }
]

Variable Substitution

Use ${step_N.field} syntax to reference results from previous steps:

**Syntax:**

${step_1.id} - ID from step 1
${step_1.name} - Name field from step 1 result
${step_1.result.data.email} - Nested field access

**Nested field access:**

# Step 1 returns: {"result": {"data": {"id": "123", "email": "test@example.com"}}}
# Step 2 can use: "${step_1.result.data.id}" → "123"

**Supported patterns:**

Simple field: ${step_1.id}
Nested field: ${step_1.result.data.email}
Multiple references: "Account: ${step_1.name} - ${step_1.billing_state}"

Proposal Pause and Resume

Workflows pause when intern agents require approval:

**Flow:**

Intern agent executes workflow
Workflow executes until write operation (Level 3+)
Proposal created for write operation
Workflow pauses with status: "paused"
Autonomous supervisor approves proposal
Workflow resumes from paused step

**Resume behavior:**

resume_from_step parameter specifies where to continue
Previous step results preserved in chain_context
Workflow continues until completion or next proposal

**Example:**

# Initial execution
result = await executor.execute_multi_step_workflow(
    tenant_id="tenant-123",
    agent_id="intern-agent-789",
    workflow_steps=[...]
)
# Returns: {"status": "paused", "steps_completed": 2, "proposal_id": "..."}

# After approval, resume from step 3
result = await executor.execute_multi_step_workflow(
    tenant_id="tenant-123",
    agent_id="intern-agent-789",
    workflow_steps=[...],
    resume_from_step=3
)
# Returns: {"status": "completed", "steps_completed": 4}

Error Handling

**Fail-fast behavior:**

Workflow stops immediately on error
Error logged to audit trail
WorldModel records failure for learning
No rollback (operations are idempotent when possible)

**Error response:**

{
    "workflow_id": "...",
    "status": "failed",
    "steps_completed": 2,
    "total_steps": 4,
    "results": [
        {"step": 1, "status": "success", "result": {...}},
        {"step": 2, "status": "error", "error": "Integration not found"}
    ]
}

API Reference

Execute Integration Operation

**Endpoint:** POST /api/agent/integration/execute

**Description:** Execute a single integration operation on behalf of an AI agent

**Request:**

{
  "connector_id": "slack",
  "operation_name": "send_message",
  "arguments": {
    "channel": "#general",
    "text": "Hello from AI agent",
    "agent_id": "agent-uuid-here"
  },
  "workspace_id": "optional-workspace-id"
}

**Request Fields:**

connector_id (string, required): Integration identifier (e.g., "slack", "salesforce")
operation_name (string, required): Operation to execute (e.g., "send_message")
arguments (object, required): Operation parameters
agent_id (string, required): Agent UUID for governance
Additional operation-specific parameters
workspace_id (string, optional): Workspace UUID for multi-tenant isolation

**Response (Success):**

{
  "status": "success",
  "result": {
    "message_id": "123456",
    "timestamp": "2026-03-28T10:00:00Z",
    "channel": "#general"
  }
}

**Response (Proposal Created):**

{
  "status": "proposal_created",
  "proposal_id": "uuid-123",
  "reason": "Intern agent requires human approval"
}

**Response (Error):**

{
  "status": "error",
  "error": "Agent maturity insufficient for this operation"
}

**Status Codes:**

200 OK: Operation executed or proposal created
400 Bad Request: Invalid parameters or governance denied
404 Not Found: Tenant or integration not found
429 Too Many Requests: Rate limit exceeded

List Available Operations

**Endpoint:** GET /api/agent/integration/operations

**Description:** List available integration operations for the current tenant

**Parameters:**

connector_id (string, optional): Filter operations by connector

**Request Examples:**

GET /api/agent/integration/operations?connector_id=slack
GET /api/agent/integration/operations

**Response:**

{
  "connector_id": "slack",
  "operations": [
    {
      "name": "send_message",
      "description": "Send a message to a Slack channel",
      "parameters": {
        "channel": {
          "type": "string",
          "required": true,
          "description": "Channel name or ID"
        },
        "text": {
          "type": "string",
          "required": true,
          "description": "Message text"
        },
        "thread_ts": {
          "type": "string",
          "required": false,
          "description": "Thread timestamp"
        }
      },
      "complexity": 3,
      "required_maturity": "supervised"
    },
    {
      "name": "get_channels",
      "description": "List all channels",
      "parameters": {},
      "complexity": 1,
      "required_maturity": "student"
    }
  ]
}

Execute Multi-Step Workflow

**Endpoint:** POST /api/agent/integration/workflow

**Description:** Execute workflow with multiple integration operations

**Request:**

{
  "workflow_steps": [
    {
      "connector": "salesforce",
      "operation": "get_account",
      "args": { "id": "001xx000003" }
    },
    {
      "connector": "slack",
      "operation": "send_message",
      "args": {
        "channel": "#sales",
        "text": "Account: ${step_1.name}"
      }
    }
  ],
  "agent_id": "agent-uuid",
  "resume_from_step": 1
}

**Request Fields:**

workflow_steps (array, required): List of workflow steps
connector (string, required): Integration identifier
operation (string, required): Operation name
args (object, required): Operation arguments with variable substitution
agent_id (string, required): Agent UUID
resume_from_step (integer, optional): Step number to resume from (default: 1)

**Response:**

{
  "workflow_id": "uuid-123",
  "status": "completed",
  "steps_completed": 2,
  "total_steps": 2,
  "results": [
    {
      "step": 1,
      "status": "success",
      "result": { "id": "001xx000003", "name": "Acme Corp" },
      "connector": "salesforce",
      "operation": "get_account"
    },
    {
      "step": 2,
      "status": "success",
      "result": { "message_id": "123456" },
      "connector": "slack",
      "operation": "send_message"
    }
  ]
}

**Status Values:**

completed: All steps executed successfully
paused: Workflow paused awaiting proposal approval
failed: Workflow stopped due to error

Developer Guide

Using EntitySkillExecutor

**Import and initialize:**

from core.entity_skill_executor import get_entity_skill_executor

executor = get_entity_skill_executor()

**Execute single operation:**

result = await executor.execute_integration_skill(
    tenant_id="tenant-uuid",
    agent_id="agent-uuid",
    connector_id="slack",
    operation_name="send_message",
    arguments={
        "channel": "#general",
        "text": "Hello from AI agent"
    },
    workspace_id="optional-workspace"
)

if result["status"] == "success":
    print(f"Success: {result['result']}")
elif result["status"] == "proposal_created":
    print(f"Proposal {result['proposal_id']} pending approval")
else:
    print(f"Error: {result['error']}")

**Execute multi-step workflow:**

result = await executor.execute_multi_step_workflow(
    tenant_id="tenant-uuid",
    agent_id="agent-uuid",
    workflow_steps=[
        {
            "connector": "salesforce",
            "operation": "get_account",
            "args": {"id": "123"}
        },
        {
            "connector": "slack",
            "operation": "send_message",
            "args": {
                "channel": "#sales",
                "text": "Account: ${step_1.name}"
            }
        }
    ]
)

print(f"Workflow {result['workflow_id']}: {result['status']}")
print(f"Steps completed: {result['steps_completed']}/{result['total_steps']}")

**Resume paused workflow:**

# After proposal approval
result = await executor.execute_multi_step_workflow(
    tenant_id="tenant-uuid",
    agent_id="agent-uuid",
    workflow_steps=[...],  # Same steps as original
    resume_from_step=3  # Resume from step 3
)

MCP Tool Registration

Integration operations are automatically registered as MCP tools for AI agent discovery:

from integrations.mcp_service import MCPService

mcp_service = MCPService()

# Register integration tools for tenant
await mcp_service.register_integration_tools(tenant_id="tenant-uuid", db=db)

# Get available tools
tools = await mcp_service.get_available_tools()

# Execute tool
result = await mcp_service.execute_integration_tool(
    tool_name="slack_send_message",
    arguments={"channel": "#general", "text": "Hello"},
    context={"tenant_id": "tenant-uuid", "agent_id": "agent-uuid"}
)

**Tool naming convention:** {connector_id}_{operation_name}

Examples:

slack_send_message
salesforce_create_lead
hubspot_update_deal

Custom Integration Service

**Implement execute_operation interface:**

from core.integration_registry import IntegrationService

class MyIntegrationService(IntegrationService):
    async def execute_operation(
        self,
        operation_name: str,
        arguments: Dict[str, Any],
        context: Dict[str, Any]
    ) -> Dict[str, Any]:
        """
        Execute operation with governance context.

        Args:
            operation_name: Operation to execute
            arguments: Operation parameters
            context: Tenant/agent/workspace context

        Returns:
            Dict with operation result
        """
        tenant_id = context["tenant_id"]
        agent_id = context["agent_id"]

        # Execute operation
        if operation_name == "create_item":
            return await self._create_item(arguments, tenant_id)
        elif operation_name == "get_item":
            return await self._get_item(arguments, tenant_id)
        else:
            return {"error": f"Unknown operation: {operation_name}"}

    def get_operations(self) -> List[Dict[str, Any]]:
        """Return list of available operations with metadata."""
        return [
            {
                "name": "create_item",
                "description": "Create a new item",
                "parameters": {
                    "name": {"type": "string", "required": True},
                    "value": {"type": "integer", "required": True}
                },
                "complexity": 3
            },
            {
                "name": "get_item",
                "description": "Get item by ID",
                "parameters": {
                    "id": {"type": "string", "required": True}
                },
                "complexity": 1
            }
        ]

**Register in IntegrationRegistry:**

# Service auto-discovered via importlib
# File: backend-saas/integrations/my_integration_service.py
# Class: MyIntegrationService
# Connector ID: "my_integration" (derived from filename)

Troubleshooting

"Agent maturity insufficient for this operation"

**Cause:** Agent maturity level too low for operation complexity

**Solutions:**

For read operations: Use student or higher agent
For write operations: Use supervised or higher agent
For delete operations: Use autonomous agent only
Promote agent maturity level via graduation system

**Example:**

Error: Agent 'student-agent-123' not allowed to execute 'slack.send_message'
Required maturity: supervised
Current maturity: student

**Fix:** Upgrade agent to supervised maturity or use different agent

"Proposal created" when expecting execution

**Cause:** Intern agents cannot execute directly; they create proposals

**Solutions:**

Wait for autonomous supervisor approval
Use supervised/autonomous agent for direct execution
Approve proposal manually via API

**Example:**

result = await executor.execute_integration_skill(...)
# Returns: {"status": "proposal_created", "proposal_id": "..."}

# Option 1: Wait for approval
# (Autonomous supervisor approves automatically)

# Option 2: Approve manually
await proposal_service.approve_proposal(
    proposal_id=result["proposal_id"],
    tenant_id=tenant_id,
    approver_id="autonomous-supervisor-id",
    reason="Approved operation",
    confidence=0.9
)

"Integration not found or not configured"

**Cause:** Integration not active for tenant or missing credentials

**Solutions:**

Verify integration is active in tenant settings
Check OAuth token or API key is configured
Verify connector_id is correct
Check tenant has integration enabled

**Debug steps:**

from core.models import TenantIntegration

integration = db.query(TenantIntegration).filter(
    TenantIntegration.tenant_id == tenant_id,
    TenantIntegration.connector_id == "slack",
    TenantIntegration.is_active == True
).first()

if not integration:
    print("Integration not found or inactive")
else:
    print(f"Integration found: {integration.credential_type}")
    # Check credentials based on credential_type

"Cross-tenant access denied"

**Cause:** Attempting to access another tenant's integration

**Solutions:**

Verify tenant_id in request matches integration owner
Check agent belongs to correct tenant
Ensure proper authentication

**Security:** This is a security feature - do not bypass

"Variable substitution failed"

**Cause:** Invalid ${step_N.field} reference in workflow

**Solutions:**

Verify step number exists
Check field name exists in step result
Use correct nested field syntax

**Example:**

# Step 1 returns: {"id": "123", "name": "Acme"}

# Correct references:
"${step_1.id}" → "123"
"${step_1.name}" → "Acme"

# Incorrect references:
"${step_2.id}" → Error (step 2 doesn't exist)
"${step_1.email}" → Error (field doesn't exist)
"{$step_1.id}" → Error (wrong syntax)

**Debug:**

# Check step results
for result in workflow_results["results"]:
    print(f"Step {result['step']}: {result.get('result', {})}")

"Rate limit exceeded"

**Cause:** Too many integration operations in short time

**Solutions:**

Wait before retrying (rate limit resets per window)
Implement exponential backoff
Upgrade plan for higher limits
Cache results to reduce redundant calls

**Rate limits by tier:**

Free: 50 operations/day
Solo: 500 operations/day
Team: 5000 operations/day
Enterprise: Custom

Examples

Example 1: Student Agent Reads Salesforce Account

**Scenario:** Student agent retrieves account details (read-only operation)

from core.entity_skill_executor import get_entity_skill_executor

executor = get_entity_skill_executor()

# Student agent can read
result = await executor.execute_integration_skill(
    tenant_id="tenant-123",
    agent_id="student-agent-456",
    connector_id="salesforce",
    operation_name="get_account",
    arguments={"id": "001xx000003"}
)

# Returns: {"status": "success", "result": {...}}
print(f"Account: {result['result']['name']}")

**Output:**

{
  "status": "success",
  "result": {
    "id": "001xx000003",
    "name": "Acme Corp",
    "billing_state": "CA",
    "type": "Customer"
  }
}

**Governance:** Level 1 operation (read-only) - Student allowed ✓

Example 2: Intern Agent Creates Proposal

**Scenario:** Intern agent attempts to send Slack message (write operation)

from core.entity_skill_executor import get_entity_skill_executor

executor = get_entity_skill_executor()

# Intern agent creates proposal for write
result = await executor.execute_integration_skill(
    tenant_id="tenant-123",
    agent_id="intern-agent-789",
    connector_id="slack",
    operation_name="send_message",
    arguments={
        "channel": "#general",
        "text": "Hello from intern agent"
    }
)

# Returns: {"status": "proposal_created", "proposal_id": "..."}
print(f"Proposal {result['proposal_id']} pending approval")

**Output:**

{
  "status": "proposal_created",
  "proposal_id": "uuid-123",
  "reason": "Intern agent requires human approval"
}

**Workflow:**

Intern agent blocked from direct execution
Proposal created for autonomous supervisor
Supervisor reviews and approves
Operation executes after approval

**Governance:** Level 3 operation (write) - Intern requires approval ✓

Example 3: Supervised Agent Executes Write

**Scenario:** Supervised agent creates Salesforce lead

from core.entity_skill_executor import get_entity_skill_executor

executor = get_entity_skill_executor()

# Supervised agent can write
result = await executor.execute_integration_skill(
    tenant_id="tenant-123",
    agent_id="supervised-agent-456",
    connector_id="salesforce",
    operation_name="create_lead",
    arguments={
        "first_name": "John",
        "last_name": "Doe",
        "email": "john.doe@example.com",
        "company": "Acme Corp"
    }
)

# Returns: {"status": "success", "result": {...}}
print(f"Lead created: {result['result']['lead_id']}")

**Output:**

{
  "status": "success",
  "result": {
    "lead_id": "003xx000001",
    "created": true,
    "timestamp": "2026-03-28T10:00:00Z"
  }
}

**Governance:** Level 3 operation (write) - Supervised allowed ✓

Example 4: Autonomous Agent Executes Delete

**Scenario:** Autonomous agent deletes Salesforce lead

from core.entity_skill_executor import get_entity_skill_executor

executor = get_entity_skill_executor()

# Autonomous agent can delete
result = await executor.execute_integration_skill(
    tenant_id="tenant-123",
    agent_id="autonomous-agent-999",
    connector_id="salesforce",
    operation_name="delete_lead",
    arguments={"id": "003xx000001"}
)

# Returns: {"status": "success", "result": {...}}
print(f"Lead deleted")

**Output:**

{
  "status": "success",
  "result": {
    "deleted": true,
    "lead_id": "003xx000001"
  }
}

**Governance:** Level 4 operation (delete) - Autonomous only ✓

Example 5: Multi-Step Workflow with Variable Substitution

**Scenario:** Read Salesforce account → Notify in Slack → Create HubSpot deal

from core.entity_skill_executor import get_entity_skill_executor

executor = get_entity_skill_executor()

# Workflow: Read account → Notify → Create deal
result = await executor.execute_multi_step_workflow(
    tenant_id="tenant-123",
    agent_id="supervised-agent-456",
    workflow_steps=[
        {
            "connector": "salesforce",
            "operation": "get_account",
            "args": {"id": "001xx000003"}
        },
        {
            "connector": "slack",
            "operation": "send_message",
            "args": {
                "channel": "#sales",
                "text": "Account: ${step_1.name} (${step_1.billing_state})"
            }
        },
        {
            "connector": "hubspot",
            "operation": "create_deal",
            "args": {
                "deal_name": "${step_1.name}",
                "amount": 50000,
                "stage": "prospecting"
            }
        }
    ]
)

# Returns: {"status": "completed", "steps_completed": 3}
print(f"Workflow {result['workflow_id']} completed")

**Output:**

{
  "workflow_id": "uuid-456",
  "status": "completed",
  "steps_completed": 3,
  "total_steps": 3,
  "results": [
    {
      "step": 1,
      "status": "success",
      "result": {
        "id": "001xx000003",
        "name": "Acme Corp",
        "billing_state": "CA"
      }
    },
    {
      "step": 2,
      "status": "success",
      "result": { "message_id": "123456" }
    },
    {
      "step": 3,
      "status": "success",
      "result": { "deal_id": "deal-789" }
    }
  ]
}

**Variable substitution:**

Step 2 text: "Account: ${step_1.name} (${step_1.billing_state})" → "Account: Acme Corp (CA)"
Step 3 deal_name: "${step_1.name}" → "Acme Corp"

Example 6: Intern Workflow with Proposal Pause

**Scenario:** Intern agent executes multi-step workflow with write operation

from core.entity_skill_executor import get_entity_skill_executor

executor = get_entity_skill_executor()

# Intern workflow pauses at write operation
result = await executor.execute_multi_step_workflow(
    tenant_id="tenant-123",
    agent_id="intern-agent-789",
    workflow_steps=[
        {
            "connector": "salesforce",
            "operation": "get_account",
            "args": {"id": "001xx000003"}
        },
        {
            "connector": "slack",
            "operation": "send_message",
            "args": {"channel": "#sales", "text": "Account: ${step_1.name}"}
        }
    ]
)

# Returns: {"status": "paused", "steps_completed": 1, "proposal_id": "..."}
print(f"Workflow paused at step 2")
print(f"Proposal {result['proposal_id']} pending approval")

**Output (initial):**

{
  "workflow_id": "uuid-789",
  "status": "paused",
  "steps_completed": 1,
  "total_steps": 2,
  "results": [
    {
      "step": 1,
      "status": "success",
      "result": { "id": "001xx000003", "name": "Acme Corp" }
    }
  ]
}

**After approval:**

# Resume from step 2
result = await executor.execute_multi_step_workflow(
    tenant_id="tenant-123",
    agent_id="intern-agent-789",
    workflow_steps=[...],  # Same steps
    resume_from_step=2
)

# Returns: {"status": "completed", "steps_completed": 2}

**Output (after approval):**

{
  "workflow_id": "uuid-789",
  "status": "completed",
  "steps_completed": 2,
  "total_steps": 2,
  "results": [
    {
      "step": 1,
      "status": "success",
      "result": { "id": "001xx000003", "name": "Acme Corp" }
    },
    {
      "step": 2,
      "status": "success",
      "result": { "message_id": "123456" }
    }
  ]
}

**Workflow:**

Step 1 (read): Intern executes directly → Success
Step 2 (write): Intern creates proposal → Paused
Supervisor approves proposal
Resume workflow from step 2 → Success

Security Considerations

Multi-Tenant Isolation

**Database layer:**

ALL queries filter by tenant_id
IntegrationRegistry enforces tenant filtering
Credentials scoped per tenant

**Service layer:**

Per-tenant service instances
Cache keys include tenant_id
No cross-tenant credential access

**Example:**

# WRONG: Cross-tenant access
service = await registry.get_service_instance("slack", tenant_id="tenant-a")
# Use service to access tenant-b integrations (BLOCKED)

# CORRECT: Tenant-scoped access
service = await registry.get_service_instance("slack", tenant_id="tenant-a")
# Only access tenant-a integrations

Governance Enforcement

**All operations checked:**

# Before every integration operation
governance_check = governance_service.can_perform_action(
    agent_id=agent_id,
    action_type=f"integration_{connector_id}_{operation_name}",
    tenant_id=tenant_id,
    db=db
)

if not governance_check["allowed"]:
    return {"status": "error", "error": governance_check["reason"]}

**Maturity levels:**

Student: Read only
Intern: Read + propose
Supervised: Read + write (no delete)
Autonomous: All operations

Audit Trail

**All actions logged:**

await audit_service.log_integration_action(
    tenant_id=tenant_id,
    agent_id=agent_id,
    connector_id=connector_id,
    operation_name=operation_name,
    outcome="success",  # or "error", "blocked"
    reason=None,
    metadata={
        "arguments": str(arguments)[:200],
        "agent_maturity": agent.status,
        "execution_time_ms": 123.45
    }
)

**Compliance:**

Full audit trail for all operations
Agent maturity tracked
Governance check results logged
Execution times recorded

Rate Limiting

**Per-tenant limits:**

Prevents abuse and runaway workflows
Enforced via AbuseProtectionService
Returns 429 when exceeded

**Plan limits:**

Free: 50/day
Solo: 500/day
Team: 5000/day
Enterprise: Custom

Core Services

**EntitySkillExecutor** - backend-saas/core/entity_skill_executor.py
execute_integration_skill() - Single operation execution
execute_multi_step_workflow() - Multi-step workflows
_inject_workflow_context() - Variable substitution

**IntegrationRegistry** - backend-saas/core/integration_registry.py
get_service_instance() - Service discovery
Per-tenant caching
Dynamic service loading

**AgentGovernanceService** - backend-saas/core/agent_governance_service.py
can_perform_action() - Maturity checks
ACTION_COMPLEXITY mapping
Permission enforcement

**ProposalService** - backend-saas/core/proposal_service.py
create_proposal() - Create intern proposals
approve_proposal() - Approve and execute
Proposal workflow management

**WorldModel** - backend-saas/core/agent_world_model.py
record_experience() - Agent learning
recall_experiences() - Pattern recognition
Semantic search

API Routes

**Agent Integration Routes** - backend-saas/api/routes/agent_integration_routes.py
POST /api/agent/integration/execute - Execute operation
GET /api/agent/integration/operations - List operations
POST /api/agent/integration/workflow - Execute workflow

Documentation

**Integration Registry** - backend-saas/docs/INTEGRATION_REGISTRY.md
**Outflow Wrappers** - backend-saas/docs/OUTFLOW_WRAPPERS.md
**Agent Governance** - backend-saas/docs/AGENT_GOVERNANCE.md
**Multi-Tenancy** - backend-saas/docs/MULTI_TENANCY.md

Phase 204 Implementation

**Phase 204 Research** - .planning/phases/204-agent-step-integration/204-RESEARCH.md
**Phase 204 Plans** - .planning/phases/204-agent-step-integration/204-*-PLAN.md
**Phase 204 Summaries** - .planning/phases/204-agent-step-integration/204-*-SUMMARY.md

Appendix

Maturity Progression

**Graduation path:**

Student → Intern: Complete read-only tasks successfully
Intern → Supervised: Proposals approved consistently
Supervised → Autonomous: High success rate, low error rate

**Readiness formula:**

readiness = (zero_intervention_ratio * 0.4) +
            (avg_constitutional_score * 0.3) +
            (avg_confidence_score * 0.2) +
            (success_rate * 0.1)

**Thresholds:**

Student → Intern: 70%
Intern → Supervised: 80%
Supervised → Autonomous: 95%

Integration Coverage

**Current integrations (39+):**

Communication: Slack, Discord, Telegram
CRM: Salesforce, HubSpot, Pipedrive
Project Management: Asana, Trello, Jira, Monday.com
Email: Gmail, Outlook, SendGrid
Storage: Google Drive, Dropbox, OneDrive
Development: GitHub, GitLab, Bitbucket
Support: Zendesk, Intercom, Freshdesk
E-commerce: Shopify, Stripe, WooCommerce
Productivity: Notion, Airtable, Google Sheets

**Adding new integrations:**

Implement IntegrationService interface
Add execute_operation() method
Add get_operations() method
Register via IntegrationRegistry (auto-discovery)

Performance Metrics

**Execution times:**

Single operation: 100-500ms (average)
Multi-step workflow: 200ms \* number of steps
Governance check: <10ms
Proposal creation: 50ms

**Cache hit rates:**

Service instance cache: 95%+
Operation metadata cache: 98%+
Tenant integration cache: 90%+

**SLA targets:**

100-step workflow: <2 seconds
Single operation: <1 second
Governance check: <50ms

Error Codes

Code	Description	Solution
`MATURITY_INSUFFICIENT`	Agent maturity too low	Upgrade agent maturity
`INTEGRATION_NOT_FOUND`	Integration not configured	Enable integration for tenant
`CREDENTIALS_MISSING`	No OAuth token or API key	Configure credentials
`RATE_LIMIT_EXCEEDED`	Too many requests	Wait or upgrade plan
`PROPOSAL_REQUIRED`	Intern agent needs approval	Wait for supervisor approval
`VARIABLE_SUBSTITUTION_FAILED`	Invalid step reference	Check step number and field name
`OPERATION_NOT_SUPPORTED`	Operation doesn't exist	Check operation name
`CROSS_TENANT_ACCESS_DENIED`	Security violation	Check tenant_id and agent ownership

---

**Version:** 1.0.0

**Last Updated:** 2026-03-28

**Maintained By:** Phase 204 Implementation Team

Agent Integration Guide

Overview

Agent Maturity Levels

Student Agents

Intern Agents

Supervised Agents

Autonomous Agents

Integration Operations

Available Operations

Slack

Salesforce

HubSpot

Asana

Gmail

Operation Complexity Levels

Multi-Step Workflows

Workflow Definition

Variable Substitution

Proposal Pause and Resume

Error Handling

API Reference

Execute Integration Operation

List Available Operations

Execute Multi-Step Workflow

Developer Guide

Using EntitySkillExecutor

MCP Tool Registration

Custom Integration Service

Troubleshooting

"Agent maturity insufficient for this operation"

"Proposal created" when expecting execution

"Integration not found or not configured"

"Cross-tenant access denied"

"Variable substitution failed"

"Rate limit exceeded"

Examples

Example 1: Student Agent Reads Salesforce Account

Example 2: Intern Agent Creates Proposal

Example 3: Supervised Agent Executes Write

Example 4: Autonomous Agent Executes Delete

Example 5: Multi-Step Workflow with Variable Substitution

Example 6: Intern Workflow with Proposal Pause

Security Considerations

Multi-Tenant Isolation

Governance Enforcement

Audit Trail

Rate Limiting

Further Reading

Core Services

API Routes

Documentation

Phase 204 Implementation

Appendix

Maturity Progression

Integration Coverage

Performance Metrics

Error Codes